//main process loop
//
#include "ax_common.h"
#include <xnt_event.h>
#include <xnt_timer.h>
#include "netd.h"
#include "conn_mgr.h"

int init_event_drv();

static int _process_init = 0;
static ListenMgr g_listener;
static ConnectMgr g_outer_conn;
static ConnectMgr g_inner_conn;


void process_client_recv(int vfd);
void process_inner_recv(int vfd);
void process_timer(int handle);
int parse_recv_packet(AxConnection *c, int* msg_type);
void dispatch_command(int vfd, unsigned char msg_type, char *buf, int cmd_len);
void on_inner_send(xnt_event_t* ev, int fd);


//message router=================================
//message will route to inner connect according to msg_type filter

#define INNER_IDENT_LEN 32

typedef struct inner_conn_entry_t
{
	int vfd;    //vfd in ConnMgr
	int msg_filter[8];   //message filter bitmap
	char name[ INNER_IDENT_LEN ];    //inner process identity name 
	struct inner_conn_entry_t* next; 
} inner_conn_entry_t;


struct inner_conn_entry_t* g_inner_router = NULL;

#define MSG_FILTER_CLEAR  0
#define MSG_FILTER_SET    1
#define MSG_FILTER_RESET  2
void set_msg_filter(inner_conn_entry_t* c, int op, int n)
{
	int i;
	if(op == MSG_FILTER_RESET) {
		for(i=0; i<8; i++) c->msg_filter[i] = 0;
	} else if (op == MSG_FILTER_SET) {
		if (n < 0 || n > 255) return;
		c->msg_filter[n/32] |= 0x01 << (n%32);
	} else if (op == MSG_FILTER_CLEAR) {
		c->msg_filter[n/32] &= ~(0x01 << (n%32)); 	
	}
}

#define GET_MSG_FILTER_BIT(x, n) (((x)->msg_filter[(n)/32] >> ((n)%32)) & 0x01)

void free_inner_conn(int vfd)
{
	inner_conn_entry_t* q;
	inner_conn_entry_t* cnode;
	//client socket fail
	g_inner_conn.free_connect(vfd);
	//free router table
	cnode = g_inner_router;
	if (cnode->vfd == vfd) {
		netd_log(LOG_CRITICAL, "inner connect close. vfd=%d, ident=%s", vfd, cnode->name);
		free(cnode);	
		g_inner_router = NULL;
		return;
	} 

	q = cnode->next;
	while(q != NULL) {
		if (q->vfd == vfd) {
			cnode->next = q->next;	
			netd_log(LOG_CRITICAL, "inner connect close. vfd=%d, ident=%s", vfd, cnode->name);
			free(q);	
			break;
		}
		cnode = q;
		q = q->next;
	}
}

//main loop
void dispatch()
{
	init_net();
	while(1) {
		xnt_process();
	}
}

//client data in
void on_outer_read(xnt_event_t *ev, int fd)
{
	int ret, vfd;
	AxConnection *c;
	//64 bit machine should use long to convert
	vfd = (long)ev->data;
	c = g_outer_conn.get_connect(vfd);
	if (c->fd != fd) {
		netd_log(LOG_CRITICAL, "outer read event. fd inconsist");
		return;
	}
	ret = c->read();	
	if (ret < 0 && (errno == EINTR || errno == EAGAIN)) {
		return;  //no data, try next time
	}
	if (ret <= 0) {
		//client socket fail
		netd_log(LOG_NORMAL, "outer client closed. vfd=%d, fd=%d, handle=%d, addr=%s:%d", vfd, fd, c->handle, c->remote_ip, c->remote_port);
		g_outer_conn.free_connect(vfd);
		return;
	}
	process_client_recv(vfd);
}

//inner read
void on_inner_read(xnt_event_t *ev, int fd)
{
	int ret, vfd;
	AxConnection *c;
	//64 bit machine should use long to convert
	vfd = (long)ev->data;
	c = g_inner_conn.get_connect(vfd);
	if (c->fd != fd) {
		netd_log(LOG_CRITICAL, "inner event fd inconsist. vfd=%d", vfd);
		return;
	}
	ret = c->read();	
	if (ret < 0 && (errno == EINTR || errno == EAGAIN)) {
		return;  //no data, try next time
	}
	if (ret <= 0) {
		free_inner_conn(vfd);
		return;
	}
		
	process_inner_recv(vfd);
}

void on_listen_accept(xnt_event_t *ev, int fd)
{
	int listen_type;
	AxListen* ls;
	AxConnection *c;
	int client_fd;
	int vfd;
	int client_type;
	struct sockaddr_in client_addr;
	ConnectMgr *cmgr;

	socklen_t length;
	listen_type = (long)ev->data;	
	ls = g_listener.get_listener(listen_type);

	if (ls == NULL)  {
		netd_log(LOG_CRITICAL, "read event error. listen_type=%d", listen_type);
		return;
	}
	if (ls->_fd != fd) {
		netd_log(LOG_CRITICAL, "listen fd inconsist. ls->_fd=%d, fd=%d, type=%d", ls->_fd, fd, listen_type);
		xnt_del_event(ev, fd, XNT_EV_READ);
		return;
	}

	length = sizeof(struct sockaddr_in);
	client_fd = ::accept(ls->_fd, (struct sockaddr *)&client_addr, (socklen_t *)&length);
	
	if (listen_type == ListenMgr::OUTER_LISTEN) {
		cmgr = &g_outer_conn;		
		client_type = OUTER_CLIENT;
	} else if (listen_type == ListenMgr::INNER_LISTEN) {
		client_type = INNER_CLIENT;
		cmgr = &g_inner_conn;
	} else {
		return;
	}

	vfd = cmgr->alloc_connect(client_fd, client_type);	
	if (vfd < 0) {
		netd_log(LOG_CRITICAL, "connection is full. reject income connection. client_type=%d", client_type);
		::close(client_fd);
		return;	
	}
	ax_set_nonblock(client_fd);
	c = cmgr->get_connect(vfd);
	c->set_addr(&client_addr);
	if (client_type == OUTER_CLIENT) {
		c->set_handler(XNT_EV_READ, on_outer_read);
		netd_log(LOG_NORMAL, "accept outer client. vfd=%d, fd=%d, addr=%s:%d", vfd, client_fd, c->remote_ip, c->remote_port);
	} else {
		inner_conn_entry_t* cnode;
		c->set_handler(XNT_EV_READ, on_inner_read);
		cnode = (inner_conn_entry_t*)malloc(sizeof(inner_conn_entry_t));
		cnode->vfd = vfd;
		cnode->next = g_inner_router;
		memset(cnode->name, 0, INNER_IDENT_LEN);
		set_msg_filter(cnode, MSG_FILTER_RESET, 0); 
		g_inner_router->next = cnode;
		netd_log(LOG_NORMAL, "accept inner client. vfd=%d, fd=%d, addr=%s:%d", vfd, client_fd, c->remote_ip, c->remote_port);
	}
}

//initialize netword data structure
void init_net()
{
	xnt_conf_t cf;
	AxListen *ls;
	if (_process_init != 0) return;

	cf.timetick = 1000 / NetdConfig::fps;
	cf.max_fd = NetdConfig::max_connect;
	if (xnt_init(&cf) != 0) {
		netd_abort("event driver init fail");
		return;
	}

	_process_init = 1;	
	ls = g_listener.get_listener(ListenMgr::OUTER_LISTEN);
	AX_ASSERT(ls != NULL);
	ls->listen(NetdConfig::out_port, 5);
	ls->rev.handler = on_listen_accept;
	ls->rev.data = (void*)ListenMgr::OUTER_LISTEN;
	xnt_add_event(&ls->rev, ls->_fd, XNT_EV_READ);
	
	ls = g_listener.get_listener(ListenMgr::INNER_LISTEN);
	AX_ASSERT(ls != NULL);
	ls->listen(NetdConfig::in_port, 5);
	ls->rev.handler = on_listen_accept;
	ls->rev.data = (void*)ListenMgr::OUTER_LISTEN;
	xnt_add_event(&ls->rev, ls->_fd, XNT_EV_READ);

	//init outer client
	g_outer_conn.alloc(NetdConfig::max_connect);
	g_inner_conn.alloc(100);

	//add timer
	xnt_add_timer(process_timer, 20);	
	xnt_add_timer(process_timer, 1000);
	xnt_add_timer(process_timer, 15000);
}

//server frame routine 
void process_timer(int handle)
{
	//xnt_add_timer(process_frame, 1500);
	printf("test frame ok. handle: %d\n", handle);
}


//process read data====================================

#define LEN_HEADER 3
#define LEN_VERIFY 1

void process_client_recv(int vfd)
{
	AxConnection *c;
	AxNetBuffer *pbuf;
	char *data;
	int pkt_len;
	int msg_type;
	c = g_outer_conn.get_connect(vfd);
	pbuf = c->rbuf;
	//1. decrypt
	

	//2. parse proto
	do {
		pkt_len = parse_recv_packet(c, &msg_type);
		if (pkt_len == -1) {
			//packet_error
			return;
		} else if (pkt_len == 0) {
			return;
		}

		data = pbuf->datap();
		data += LEN_HEADER;
		//send command to app
		dispatch_command(vfd, msg_type, data, pkt_len-LEN_HEADER-LEN_VERIFY);
		pbuf->pop(pkt_len);
	} while (pkt_len > 0);
}

//incomming proto
// +===============+=================+===========+=============|
// | length(2b)    | msg_type(1byte) | data(...) | verify byte |  
// +===============+=================+===========+=============|
// length include: 2b+1b+len(data)+1b
// LEN_HEADER = len(length + msg_type)
// length = length[0] << 8 | length[1]   big endian
// verify byte = (0xFF & length << 6) | (data[0] ^ data[last] ^ 0x69) & 0x1F
//
// @return packet len       -1  packet error
//         msg_type
int parse_recv_packet(AxConnection *c, int* msg_type)
{
	int vbyte;
	AxNetBuffer *pbuf = c->rbuf;	
	char *p = pbuf->datap();
	unsigned int recv_len, pkt_len, cmd_len;
	
	pkt_len = p[0] << 8 | p[1];
	recv_len = pbuf->size();
	if (recv_len < pkt_len) return 0;

	cmd_len = pkt_len - LEN_HEADER - LEN_VERIFY;
	*msg_type = p[LEN_HEADER-1];
	p += LEN_HEADER;
	if (cmd_len >= 1) {
		vbyte = 0xFF & cmd_len << 6;
		vbyte |= (p[0] ^ p[cmd_len-1] ^ 0x69) & 0x3F;
	} else {
		//no data field
		vbyte = (0x07 & cmd_len) << 5;
	}
	//verify packet
	if (vbyte != p[cmd_len]) {
		pbuf->pop(pkt_len);
		return -1;
	}
	c->add_cmd_count();
	return pkt_len;
}

//send command to application
//dispatch it filter by msg_type
void dispatch_command(int vfd, unsigned char msg_type, char *buf, int cmd_len)
{
	AxConnection *c;
	inner_conn_entry_t* cnode;
	AxCmdHeader head;

	head.kind = AX_COMM_OUTER;
	head.msg_type = msg_type;
	head.handle = vfd;
	head.datalen = cmd_len;
	
	cnode = g_inner_router;
	while (cnode != NULL) {
		if (GET_MSG_FILTER_BIT(cnode, (int)msg_type) == 0) {
			cnode = cnode->next;
			continue;
		}
		c = g_inner_conn.get_connect(cnode->vfd);	
		if (c == NULL) {
			cnode = cnode->next;
			continue;
		}
		c->wbuf->push((char*)&head, sizeof(head));
		c->wbuf->push(buf, cmd_len);
		c->send_remain();
		if (c->wbuf->size() > 0) {
			c->wev.handler = on_inner_send;
			c->wev.data = (void*)cnode->vfd;
			xnt_add_event(&c->wev, c->fd, XNT_EV_WRITE);
		}	
		cnode = cnode->next;
	}
}

//inner socket recv
void process_inner_recv(int vfd)
{
	AxConnection *c;
	c = g_inner_conn.get_connect(vfd);
	
}


//inner send
void on_inner_send(xnt_event_t* ev, int fd)
{
	int ret;
	int vfd;
	AxConnection *c;
	vfd = (long)ev->data;
	c = g_inner_conn.get_connect(vfd);
	if (c == NULL) return;
	ret = c->send_remain();	
	if (ret < 0) {
		//connection fail
		free_inner_conn(vfd);	
		return;
	}
	if (c->wbuf->size() <= 0) {
		//finish send
		xnt_del_event(&c->wev, c->fd, XNT_EV_WRITE);
	}
}

//outer client send
void on_outer_send(xnt_event_t* ev, int fd)
{
	int ret;
	int vfd;
	AxConnection *c;
	vfd = (long)ev->data;
	c = g_outer_conn.get_connect(vfd);
	if (c == NULL) return;
	ret = c->send_remain();	
	if (ret < 0) {
		g_outer_conn.free_connect(vfd);	
		return;
	}
	if (c->wbuf->size() <= 0) {
		xnt_del_event(&c->wev, c->fd, XNT_EV_WRITE);
	}
}
